package com.jjb.spider.mqtt.service.single;

import com.jjb.spider.mqtt.cache.NodeChannelHandlerContextCache;
import com.jjb.spider.mqtt.ctx.MqttChannelHandlerContext;
import com.jjb.spider.mqtt.model.DeviceTopSubscribeMessage;
import com.jjb.spider.mqtt.model.TopSubscribeMessage;
import com.jjb.spider.mqtt.service.IMqttMessageService;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Mono;

import java.util.*;
import java.util.stream.Collectors;

@Service
public class MqttSingleMessageService extends IMqttMessageService {


    @Override
    public Mono<Boolean> sendMsgByDeviceId(String deviceId, Object obj) {
        MqttChannelHandlerContext mqttChannelHandlerContext = NodeChannelHandlerContextCache.getByDeviceId(deviceId);
        if ( mqttChannelHandlerContext != null ) {
            mqttChannelHandlerContext.writeAndFlush(obj);
            return Mono.just(true);
        } else {
            return Mono.just(false);
        }
    }

    @Override
    public List<DeviceTopSubscribeMessage> getSendDeviceIdByTopic(String topics) {
        Collection<MqttChannelHandlerContext> channels = NodeChannelHandlerContextCache.getChannels();
        if( channels == null || channels.isEmpty()) {
            return null;
        } else {
            List<DeviceTopSubscribeMessage> result = new LinkedList<>();
            Iterator<MqttChannelHandlerContext> it = channels.iterator();
            while (it.hasNext()) {
                MqttChannelHandlerContext mqttChannelHandlerContext = it.next();
                List<TopSubscribeMessage> mqttTopics = mqttChannelHandlerContext.getSubscribeTopics(topics);
                if( ! (mqttTopics == null || mqttTopics.isEmpty() )) {
                    result.addAll(
                            mqttTopics.stream().map(mqttTopic -> new DeviceTopSubscribeMessage(mqttChannelHandlerContext.getDeviceId(), mqttTopic.getTopic(), mqttTopic.getQos())).collect(Collectors.toList())
                    );
                }
            }
            return result;
        }
    }

    @Override
    public Mono<Void> closeConnectHandler(MqttChannelHandlerContext ctx) {
        return Mono.empty();
    }
}
